package cn.edu.xmu.datamining.tangzk.mralgos.pagerank;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * PageRank Mapper
 * input: <long, (unnormalized pagerank, adjnodes)>
 * adjnodes format: id1, id2, id3
 * 
 * output:
 * -> <id, (original pagerank, adjnodes)>
 * -> <id, (unnormalized pagerank)
 * 
 * @version 2013-8-4
 * @author tangzk
 * @Reviewer
 * 
 */
public class PageRankMapper
		extends
		Mapper<LongWritable, Text, PageRankNodeKeyWritable, PageRankNodeValueWritable> {

	protected void map(
			LongWritable key,
			Text value,
			org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, PageRankNodeKeyWritable, PageRankNodeValueWritable>.Context context)
			throws java.io.IOException, InterruptedException {
		// format: id, pagerank, id1,id2,...
		String val = value.toString();
		String[] flds = val.split("\t");
		if (flds.length < 2) {
			context.getCounter("UserDefined", "bad records").increment(1);
			return;
		}
		int srcId = Integer.parseInt(flds[0]);
		float pagerank = Float.parseFloat(flds[1]);
		if (flds.length >= 3) {
			PageRankNodeValueWritable node = new PageRankNodeValueWritable(
					pagerank, flds[2].trim());
			context.write(new PageRankNodeKeyWritable(srcId, 0), node);

			// count the total number of nodes for random surfing
			context.getCounter(PageRankDriver.USER_DEFINED_GROUP,
					PageRankDriver.TOTAL_NODES).increment(1);

			// sum up all pagerank to normalized to one in reduce
			context.getCounter(PageRankDriver.USER_DEFINED_GROUP,
					PageRankDriver.SUM_PAGERANK_COUNTER).increment(
					(long) (pagerank * PageRankDriver.PAGERANK_BASE));

			int len = node.getAdjNodes().size();
			for (Integer destId : node.getAdjNodes()) {
				context.write(new PageRankNodeKeyWritable(destId, 1),
						new PageRankNodeValueWritable(pagerank, len));
			}
		}
	};

}
